package com.atguigu.gmall.realtime.app

import java.{lang, util}
import java.text.SimpleDateFormat
import java.util.Date

import com.alibaba.fastjson.{JSON, JSONObject}
import com.atguigu.gmall.realtime.bean.DauInfo
import com.atguigu.gmall.realtime.util.{MyEsUtil, MyKafakUtil, OffsetManager, RedisUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.elasticsearch.action.search.{SearchRequest, SearchResponse}
import org.elasticsearch.client.{RequestOptions, RestHighLevelClient}
import org.elasticsearch.search.SearchHit
import org.elasticsearch.search.builder.SearchSourceBuilder
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer

object DwDauInfoApp {

  //uv  pv  dau
  // page view   user visit    daily active user


  //0 环境
  //1 加载kafka数据
  //2 提取偏移
  //3  转换结构化对象
  //4 去重处理
  //5  拼接维度信息
  //6 写入到es中
  def main(args: Array[String]): Unit = {
    //0 环境
    val topic ="DWD_PAGE_LOG"
    val groupId = "dw_dau_app"

    val sparkConf: SparkConf = new SparkConf().setAppName("dw_dau_app").setMaster("local[4]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    //0.5 还原状态
    revertState()

    //1 加载kafka数据
    val offsetMap: Map[TopicPartition, Long] = OffsetManager.getOffset(topic, groupId)

    var inputDstream: InputDStream[ConsumerRecord[String, String]] = null
    if (offsetMap == null || offsetMap.size == 0) {
      inputDstream = MyKafakUtil.getKafkaStream(topic, ssc, groupId)
    } else {
      inputDstream = MyKafakUtil.getKafkaStream(topic, ssc, offsetMap, groupId)
    }

    //2 提取偏移
    var offsetRanges: Array[OffsetRange] = null //driver
    val inputDstreamWithOffset: DStream[ConsumerRecord[String, String]] = inputDstream.transform { rdd =>
      offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      rdd
    }

    //3  转换结构化对象
    val jsonObjDstream: DStream[JSONObject] = inputDstreamWithOffset.map { record => //ex
      val jsonString: String = record.value()
      val jsonObj: JSONObject = JSON.parseObject(jsonString)

      val ts: lang.Long = jsonObj.getLong("ts")
      val dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
      val dateTimeStr: String = dateFormat.format(new Date(ts))
      val dateTimeArr: Array[String] = dateTimeStr.split(" ")
      jsonObj.put("dt",dateTimeArr(0))
      jsonObj.put("time",dateTimeArr(1))
      jsonObj
    }
   // jsonObjDstream.print(1000)

    //4 去重处理
    //  保留用户当日访问的第一条 记录 mid



    // 维护一个当日mid访问的清单  状态 redis
    // type ?  string   set        key?   DAU:2021-12-21 value ?   mids  field/score?无
    //  写入 ？ set (? setnx)   sadd  读取？sadd        过期？ 24个小时

    val filteredDstream: DStream[JSONObject] = jsonObjDstream.mapPartitions { jsonObjItr =>
      val jsonObjList: List[JSONObject] = jsonObjItr.toList  //注意迭代器 只能使用一次  所以要提取到集合中
      println(s"过滤前: ${jsonObjList.size}条数据！")
      val filteredList: ListBuffer[JSONObject] = ListBuffer[JSONObject]()
      val jedis: Jedis = RedisUtil.getJedisClient
      for (jsonObj <- jsonObjList) {
      //redis进行自检  last_page_id 为空的保留 ，不为空直接筛掉 减少redis的工作量
       var ifNew: lang.Long=0L
        val lastPageId: String = jsonObj.getString("last_page_id")
        if(lastPageId==null||lastPageId.size==0){
          val dt: String = jsonObj.getString("dt")
          val mid: String = jsonObj.getString("mid")
          val dauKey = s"DAU:$dt"
            ifNew  = jedis.sadd(dauKey,mid)

          if (ifNew == 1L) {
            filteredList.append(jsonObj)
          }
        }
      //  println("!!!!!!!!!!!!")

      }
      jedis.close()
      println(s"过滤后: ${filteredList.size}条数据！")
      filteredList.toIterator

    }
  //  filteredDstream.print(1000)

    //5  拼接维度信息
    // 把每个外键id (地区id \用户id \商品id  ）去查询对应的数据库（redis /hbase /mysql / 外部服务)
    //
    val dauInfoDstream: DStream[DauInfo] = filteredDstream.map { jsonObj =>

      val jedis: Jedis = RedisUtil.getJedisClient
      //补充用户信息
      val uid: String = jsonObj.getString("user_id")
      val userKey=s"DIM:user_info:$uid"
      val userInfoJson: String = jedis.get(userKey)
      val gender: String = null

      val age: Long = 0L
      val ts: lang.Long = jsonObj.getLong("ts")
      if(userInfoJson!=null &&userInfoJson.size>0){
        val userJsonObj: JSONObject = JSON.parseObject(userInfoJson)

        val gender: String = userJsonObj.getString("gender")
        val birthday: String = userJsonObj.getString("birthday")
        val dateFormat = new SimpleDateFormat("yyyy-MM-dd")
        val birthdayDate: Date = dateFormat.parse(birthday)
        val birthdayTs: Long = birthdayDate.getTime
        val age: Long = ( ts-birthdayTs) /1000/3600/24/365

      }

      //地区信息
      val provinceId: String = jsonObj.getString("province_id")
      val provinceKey=s"DIM:base_province:$provinceId"

      val provinceInfoJson: String = jedis.get(provinceKey)
      val provinceJsonObj: JSONObject =null;
      val provinceName: String=null;
      val isoName: String =null;
      val iso31662: String =null;
      val areaCode: String=null;
      if(provinceInfoJson!=null &&provinceInfoJson.size>0) {
        val provinceJsonObj: JSONObject = JSON.parseObject(provinceInfoJson)
        val provinceName: String = provinceJsonObj.getString("name")
        val isoName: String = provinceJsonObj.getString("iso_code")
        val iso31662: String = provinceJsonObj.getString("iso_3166_2")
        val areaCode: String = provinceJsonObj.getString("area_code")
      }

      val mid: String = jsonObj.getString("mid")
      val user_id: String = jsonObj.getString("user_id")
      val province_id: String = jsonObj.getString("province_id")
      val channel: String = jsonObj.getString("channel")
      val is_new: String = jsonObj.getString("is_new")
      val model: String = jsonObj.getString("model")
      val operate_system: String = jsonObj.getString("operate_system")
      val version_code: String = jsonObj.getString("version_code")
      val page_id: String = jsonObj.getString("page_id")
      val page_item: String = jsonObj.getString("page_item")
      val page_item_type: String = jsonObj.getString("page_item_type")
      val during_time: Long = jsonObj.getLong("during_time")

      val dt: String = jsonObj.getString("dt")
      val timestr: String = jsonObj.getString("time")
      val timeArr: Array[String] = timestr.split(":")
      val hr: String = timeArr(0)
      jedis.close()
      DauInfo(mid,user_id,province_id,channel,is_new,model,operate_system,version_code,page_id,page_item,
        page_item_type,during_time,gender,age.toString,provinceName,isoName,iso31662,areaCode,dt,hr,ts ) //补充字段

    }
   // dauInfoDstream.print(1000)

    //TODO
    //6 写入到es中
    // 6.1  es 建索引？建索引模板
    // 6.2  写入es  1  、批量写入  2、精确一次   后置提交偏移量  driver + 幂等性保存 ？

    dauInfoDstream.foreachRDD{rdd=>
      rdd.foreachPartition{dauInfoItr=>

        val dauInfoList: List[DauInfo] = dauInfoItr.toList
        if(dauInfoList!=null&&dauInfoList.size>0){
          val dauInfoFirst: DauInfo = dauInfoList(0)
          //每个元素增加 id 保证幂等性
          val dauInfoWithIdList: List[(String, DauInfo)] = dauInfoList.map(dauInfo=>(dauInfo.mid,dauInfo))
          val dt: String = dauInfoFirst.dt
          val indexName=s"gmall0722_dau_info_$dt"
          MyEsUtil.saveBulk(dauInfoWithIdList,indexName)

        }

      }
      //c driver
      OffsetManager.saveOffset(topic,groupId,offsetRanges)


    }


    ssc.start()
    ssc.awaitTermination()

  }



  def  revertState(): Unit ={
    // 1 查询 es   当日访问过的mid 清单
//    GET gmall0722_dau_info_2021-12-25/_search
//    {
//      "_source": "mid",
//      "size": 10000
//    }
//  打开数据结果显示上限
    //   PUT /_settings
    //{
    //  "index.max_result_window" :"5000000"
    //}


    val esClient: RestHighLevelClient = MyEsUtil.getClientNew

    val dateformat = new SimpleDateFormat("yyyy-MM-dd")
    val dt: String = dateformat.format(new Date())
    val indexName=s"gmall0722_dau_info_$dt"
    val searchSourceBuilder = new SearchSourceBuilder()
    searchSourceBuilder.size(10000)
    searchSourceBuilder.fetchSource("mid",null)

    val searchRequest = new SearchRequest(indexName).source(searchSourceBuilder)
    var response: SearchResponse=null
    try{
      response  = esClient.search(searchRequest,RequestOptions.DEFAULT)
    } catch {
      case e => {
         println("不存在 索引")
         return
      }
    }
    val hits: Array[SearchHit] = response.getHits.getHits
     val midSet = new util.HashSet[String]()
    for (hit <- hits) {
      val map: util.Map[String, AnyRef] = hit.getSourceAsMap
      val mid: AnyRef = map.get("mid")
      midSet.add(  mid.asInstanceOf[String])
    }
    //2 把mid清单 写入redis 状态 即可
    // 2.1 把残留的状态清除
    val jedisClient: Jedis = RedisUtil.getJedisClient
    val dauKey=   s"DAU:$dt"
    jedisClient.del(dauKey)
    //2.2 加入
    val midArr: Array[String] = midSet.toArray( new Array[String](midSet.size()))
    jedisClient.sadd(dauKey,midArr:_*)
    jedisClient.close()


  }

}
